package org.wx.herostory.services.util.mq;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListener;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.wx.herostory.services.rank.RankService;

import java.util.List;

/**
 * @description: mq 消费者
 * @author: wangxun
 * @createDate: 2021/3/8
 * @version: 1.0
 */
@Slf4j
public class MqConsumer {
    private MqConsumer(){}

    static public void init(){
        final DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("herostory");
        // namesrv 地址
        consumer.setNamesrvAddr("test.wx.cn:9876");

        try {
            consumer.subscribe("herostory_victor","*");
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgList, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                    for (MessageExt messageExt : msgList) {
                        if(messageExt == null){
                            continue;
                        }
                        final VictorMsg victorMsg = JSONObject.parseObject(messageExt.getBody(), VictorMsg.class);
                        log.info("mq consumer 接受到消息，胜利者是{} 失败者是{}",victorMsg.getWinnerId(),victorMsg.getLoserId());
                        // 刷新排行榜
                        RankService.getInstance().refreshRank(victorMsg.getWinnerId(),victorMsg.getLoserId());
                    }
                    // 确认成功
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
            log.info("消息队列【消费者】初始化成功");
        } catch (Exception e) {
            log.error(e.getMessage(),e);
        }

    }
}
